package com.credithc.eagleeye.plugins.rabbitmq;
import com.credithc.eagleeye.agent.factory.EagleEyeAgentFactory;
import com.credithc.eagleeye.agent.util.HostUtil;
import com.credithc.eagleeye.constants.RpcTypeEnum;
import com.credithc.eagleeye.exception.EagleEyeIgnoreException;
import com.credithc.eagleeye.plugins.util.LogUtil;
import com.credithc.eagleeye.thread.EagleEyeThreadContext;
import com.credithc.eagleeye.thread.TraceInfo;
import com.credithc.eagleeye.util.MixUtil;
import com.credithc.eagleeye.util.ThrowableToStringArray;
import com.credithc.eagleeye.vo.RpcTraceInfoVo;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;

import static com.credithc.eagleeye.constants.EagleEyeConstant.HC_RPC_ID;
import static com.credithc.eagleeye.constants.EagleEyeConstant.HC_TRACE_ID;
/**
 * 描 述: <br/>
 * 作 者: zhangjin<br/>
 * 创 建：2019年10月24日<br/>
 * 版 本：v1.0.0<br>
 * <p>
 * 历 史: (版本) 作者 时间 注释 <br/>
 */
@Slf4j
public class EagleEyeRabbitListenerAdvice implements MethodInterceptor {
    private static int maxLength = 4000;
    /**
     * MethodInterceptor for {@link SimpleMessageListenerContainer.ContainerDelegate (Channel,Message)}
     */
    @Override
    public Object invoke(MethodInvocation methodInvocation) throws Throwable {
        Object resultObject = null;
        log.debug("EagleEyeRabbitListenerAdvice around start...");
        Channel channel = (Channel) methodInvocation.getArguments()[0];
        Message message = (Message) methodInvocation.getArguments()[1];

        TraceInfo traceInfo = initEagleEyeTraceInfo(message);
        RpcTraceInfoVo rpcTraceInfoVo = initRpcTraceInfoVo(traceInfo,message,channel);
        //mq为入口，下边的调用链再增加一层级
        traceInfo.addHierarchy();
        //执行调用
        long beginTime = System.currentTimeMillis();
        try {
            resultObject = methodInvocation.proceed();
            rpcTraceInfoVo.setRunTime(System.currentTimeMillis()-beginTime);
            rpcTraceInfoVo.setResult("OK");
        } catch (Exception ex) {
            if (!(ex instanceof EagleEyeIgnoreException)) {
                //ERROR
                rpcTraceInfoVo.setResult("ERROR");
                rpcTraceInfoVo.setRunTime(System.currentTimeMillis()-beginTime);
                //异常的结果
                String[] traceStr = ThrowableToStringArray.convert(ex);
                StringBuilder builder = new StringBuilder();
                for(String trace : traceStr) {
                    builder.append(trace);
                    builder.append("\n");
                }
                rpcTraceInfoVo.setResponseJson(builder.toString());
            } else {
                //OK
                rpcTraceInfoVo.setResult("OK");
            }
            throw ex;
        } finally {
            EagleEyeAgentFactory.getInstance().send2EagleEye(rpcTraceInfoVo);
            LogUtil.printSplunkLog(rpcTraceInfoVo);
            EagleEyeThreadContext.removeTraceInfo();
            MDC.remove("traceId");
            MDC.remove("rpcId");
            MDC.remove("clientIp");
        }
        log.debug("EagleEyeRabbitListenerAdvice around end...");
        return resultObject;

    }

    private TraceInfo initEagleEyeTraceInfo(Message message) {
        EagleEyeThreadContext.init();
        TraceInfo traceInfo = new TraceInfo();
        MessageProperties messageProperties = message.getMessageProperties();
        Map<String, Object> headers = messageProperties.getHeaders();
        Object hcTraceId = headers.get(HC_TRACE_ID);
        Object hcRpcId = headers.get(HC_RPC_ID);
        if (hcTraceId!=null && StringUtils.isNotBlank(hcTraceId.toString())) {
            traceInfo.setTraceId(hcTraceId.toString());
        }
        if (hcRpcId!=null && StringUtils.isNotBlank(hcRpcId.toString())) {
            traceInfo.setRpcId(hcRpcId.toString());
            traceInfo.addHierarchy();
            traceInfo.addSequenceNo();
        }
        EagleEyeThreadContext.putTraceInfo(traceInfo);
        MDC.put("traceId", traceInfo.getTraceId());
        MDC.put("rpcId", traceInfo.getRpcId());
        return traceInfo;
    }

    private RpcTraceInfoVo initRpcTraceInfoVo(TraceInfo traceInfo,Message message, Channel channel) {
        RpcTraceInfoVo rpcTraceInfoVo = new RpcTraceInfoVo();
        Date date = new Date();
        rpcTraceInfoVo.setRequestDateTime(MixUtil.ISO_DATETIME_TIME_ZONE_FORMAT_WITH_MILLIS.format(date));
        rpcTraceInfoVo.setTraceId(traceInfo.getTraceId());
        rpcTraceInfoVo.setRpcId(traceInfo.getRpcId());
        rpcTraceInfoVo.setRpcType(RpcTypeEnum.MQ.getRpcName());
        rpcTraceInfoVo.setServiceCategory("rabbitmq");
        rpcTraceInfoVo.setServiceName("rabbitmq");
        rpcTraceInfoVo.setMethodName("rabbitListener");
        rpcTraceInfoVo.setClientHost(HostUtil.getIp());
        rpcTraceInfoVo.setServerHost(channel.getConnection().toString());
        MessageProperties messageProperties = message.getMessageProperties();
        rpcTraceInfoVo.setRequestSize(messageProperties.getContentLength());
        StringBuilder requestJsonBuilder = new StringBuilder();
        requestJsonBuilder.append("exchange:" + messageProperties.getReceivedExchange() + "\n");
        requestJsonBuilder.append("routingKey:" + messageProperties.getReceivedRoutingKey() + "\n");
        requestJsonBuilder.append("queue:" + messageProperties.getConsumerQueue() + "\n");
        requestJsonBuilder.append("body:");
        if (StringUtils.isBlank(messageProperties.getContentType())) {
            try {
                requestJsonBuilder.append(fixRequestJson(new String(message.getBody(), StandardCharsets.UTF_8)));
            } catch (Exception e) {
            }
        } else if (messageProperties.getContentType().equals(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)) {
            try {
                String contentEncoding = messageProperties.getContentEncoding();
                if (contentEncoding == null) {
                    contentEncoding = "UTF-8";
                }
                requestJsonBuilder.append(fixRequestJson(new String(message.getBody(), contentEncoding)));
            } catch (UnsupportedEncodingException e) {
            }
        } else if (messageProperties.getContentType().equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
            requestJsonBuilder.append("data type object");
        } else if (messageProperties.getContentType().equals(MessageProperties.CONTENT_TYPE_BYTES)) {
            requestJsonBuilder.append("data type byte[]");
        }
        rpcTraceInfoVo.setRequestJson(requestJsonBuilder.toString());
        return rpcTraceInfoVo;
    }
    private String fixRequestJson(String messageStr) {
        if (messageStr != null && messageStr.length() > maxLength) {
            return StringUtils.substring(messageStr, 0, maxLength) + "..";
        }
        return messageStr;
    }

}
